package org.apache.nutch.scoring.webgraph.algorithms;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.ObjectWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.ToolRunner;
import org.apache.nutch.scoring.webgraph.LinkDatum;
import org.apache.nutch.scoring.webgraph.LinkRank;
import org.apache.nutch.scoring.webgraph.Node;
import org.apache.nutch.util.NutchConfiguration;

public class InDegree extends LinkRank
{
	private void initialize()
	{
		setAnalyzerMapClass(Analyzer.class);
		setAnalyzerReduceClass(Analyzer.class);
		setNormalizerMapClass(null);
		setNormalizerReduceClass(null);
		setConvergerMapClass(null);
		setConvergerReduceClass(null);
	}

	public InDegree()
	{
		initialize();
	}

	public InDegree(Configuration conf)
	{
		super(conf);
		initialize();
	}

	/**
	 * Runs a single link analysis iteration.
	 */
	private static class Analyzer implements
			Mapper<Text, Writable, Text, ObjectWritable>,
			Reducer<Text, ObjectWritable, Text, Node>
	{
		private JobConf conf;

		public void configure(JobConf conf)
		{
			this.conf = conf;
		}

		/**
		 * Convert values to ObjectWritable
		 */
		public void map(Text key, Writable value,
				OutputCollector<Text, ObjectWritable> output, Reporter reporter)
				throws IOException
		{
			ObjectWritable objWrite = new ObjectWritable();
			objWrite.set(WritableUtils.clone(value, conf));
			output.collect(key, objWrite);
		}

		/**
		 * Indegree reduce-function.
		 * 
		 */
		public void reduce(Text key, Iterator<ObjectWritable> values,
				OutputCollector<Text, Node> output, Reporter reporter)
				throws IOException
		{
			Node node = null;
			int inDegree = 0;

			while (values.hasNext())
			{
				Object value = values.next().get();
				
				if (value instanceof Node)
					node = (Node) value;
				else if (value instanceof LinkDatum)
				{
					LinkDatum link = (LinkDatum)value;

					if (link.getLinkType() == LinkDatum.INLINK)
						inDegree++;
				}
			}

			// store the score in a temporary NodeDb
			Node outNode = (Node) WritableUtils.clone(node, conf);
			LOG.info("Setting score of " + key + " to " + inDegree);
			outNode.setInlinkScore(inDegree);
			output.collect(key, outNode);
		}

		public void close() throws IOException
		{
		}
	}

	public int run(String[] args) throws Exception
	{
		getConf().setInt("link.analyze.num.iterations", 1);
		return super.run(args);
	}

	public static void main(String[] args) throws Exception
	{
		InDegree alg = new InDegree();

		int res = ToolRunner.run(NutchConfiguration.create(), alg, args);
		System.exit(res);
	}

}
